/**
 * Copyright 2011 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.google.api.adwords.starterkit.examples.enhancedads;

import com.google.api.adwords.starterkit.db.structures.KeywordParamStructure;
import com.google.api.adwords.starterkit.db.tables.AdGroupTable;
import com.google.api.adwords.starterkit.db.tables.CampaignTable;
import com.google.api.adwords.starterkit.db.tables.TextAdTable;
import com.google.api.adwords.starterkit.examples.enhancedads.db.TextAd2CatId;
import com.google.api.adwords.starterkit.examples.enhancedads.db.TextAd2ProdId;
import com.google.api.adwords.starterkit.examples.enhancedads.settings.AccountSettings;
import com.google.api.adwords.starterkit.examples.enhancedads.settings.EmailSettings;
import com.google.api.adwords.starterkit.utils.DumpExceptions;

import org.apache.log4j.Logger;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashSet;
import java.util.Iterator;

/**
 * The Class DB2AdWords process all DestUrl params stored in the local
 * DB and validates it, updating AdParam values, pause/unpausing TextAds,
 * etc.
 * 
 * @author rodrigo.f.vale@gmail.com
 */
public class DB2AdWords {
  private Connection conn;
  private Currency currency;
  private TextAdTable textAdTable;
  private AdGroupTable adGroupTable;
  private CampaignTable campaignTable;
  private TextAd2CatId textAd2CatId;
  private TextAd2ProdId textAd2ProdId;
  private static final Logger log = Logger.getLogger("StarterKit"); 
  private EmailSettings emailSettings;
  private AccountSettings account;
  
  /**
   * Instantiates a new db2adwords.
   *
   * @param conn the connectio to StarterKit DB
   * @param currency the currency object used to format float values
   */
  public DB2AdWords(Connection conn, Currency currency, 
      AccountSettings account, EmailSettings emailSettings) {
    this.conn = conn;
    this.currency = currency;
    this.emailSettings = emailSettings;
    this.account = account;
    textAdTable = new TextAdTable(conn);
    adGroupTable = new AdGroupTable(conn);
    campaignTable = new CampaignTable(conn);
    textAd2CatId = new TextAd2CatId(conn);
    textAd2ProdId = new TextAd2ProdId(conn);
  }

  /**
   * Gets the textads' localid from DB that is linked to 
   * a product in the catalog, but this product is not available in the stock 
   *
   * @return a set with the textads' localids 
   */
  private HashSet<Integer> getUnAvailableTextAds() {
    PreparedStatement prodIdTextAds;
    HashSet<Integer> stack = new HashSet<Integer>();
    try {
      prodIdTextAds = conn.prepareStatement(
          "SELECT tc.LocalId FROM TextAd2CatId tc LEFT JOIN Category ca " +
          "ON (ca.CategoryId = tc.CategoryId)  WHERE ca.CategoryId is NULL");
      ResultSet rs = prodIdTextAds.executeQuery();

      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();

      prodIdTextAds = conn.prepareStatement(
          "SELECT tp.LocalId FROM TextAd2ProdId tp LEFT JOIN Catalog ca " +
          "ON (ca.ProdId = tp.ProdId)  WHERE ca.ProdId is NULL");

      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();      
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return stack;
  }
  
  /**
   * Pause all textads in the set textAdIds.
   *
   * @param textAdIds the set with the list of localids to be paused
   */
  public void pauseTextAds(HashSet<Integer> textAdIds) {
    try {
      PreparedStatement update = conn.prepareStatement(
          "UPDATE TextAds SET Status = 'PAUSED', IsTouched = 1, " +
          "IsUpdated = 1, IsBanned = 0 WHERE LocalId IN (" + 
          getInClause(textAdIds) + ") " + "AND Status <> 'PAUSED'"
      );
      update.executeUpdate();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
  }  

  /**
   * Gets the textads linked to the catalog of products. This method will
   * return only textads linked to products in stock. 
   *
   * @return a set with localids related to TextAds in the local DB
   */
  private HashSet<Integer> getTextAdsInStock() {
    PreparedStatement prodIdTextAds;
    HashSet<Integer> stack = new HashSet<Integer>();
    try {
      // gets the TextAds' localIds from ads linked to a product
      prodIdTextAds = conn.prepareStatement(
          "SELECT DISTINCT LocalId FROM TextAd2ProdId ap, Catalog ca " +
          "WHERE ca.ProdId = ap.ProdId");
      ResultSet rs = prodIdTextAds.executeQuery();

      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();

      // gets the TextAds localIds from ads linked to a category
      prodIdTextAds = conn.prepareStatement(
          "SELECT DISTINCT LocalId FROM TextAd2CatId ac, Category ca " +
          "WHERE ca.CategoryId = ac.CategoryId");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return stack;
  }
  
  /**
   * Gets adGroupsLocalId from AdGroups with at least one active textad tied
   * to the catalog of products
   *
   * @return a set with localadgroupids 
   */
  private HashSet<Integer> getAdGroupsWithTextAdsInStock() {
    PreparedStatement select;
    HashSet<Integer> stack = new HashSet<Integer>();
    try {
      select = conn.prepareStatement(
          "SELECT DISTINCT ta.LocalAdgroupId FROM TextAds ta, " +
          "TextAd2CatId tc " +
          "WHERE tc.LocalId = ta.LocalId AND ta.Status = 'enabled'");
      ResultSet rs = select.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();

      select = conn.prepareStatement(
          "SELECT DISTINCT ta.LocalAdgroupId FROM TextAds ta, " +
          "TextAd2ProdId tp " +
          "WHERE tp.LocalId = ta.LocalId AND ta.Status = 'enabled'");
      rs = select.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
      
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return stack;
  }
  
  /**
   * Gets adGroupsLocalId from AdGroups with textad using the flag 
   * 'aw_alternative'
   *
   * @return a set with localadgroupids 
   */
  private HashSet<Integer> getAdGroupsWithAlternativeTextAds() {
    PreparedStatement select;
    HashSet<Integer> stack = new HashSet<Integer>();
    try {
      select = conn.prepareStatement(
          "SELECT DISTINCT ta.LocalAdgroupId FROM TextAds ta, Schedule sc " +
          "WHERE sc.LocalId = ta.LocalId");
      ResultSet rs = select.executeQuery();

      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return stack;
  }
  
  /**
   * Gets TextAds using the flag 'aw_alternative' from a list of AdGroups  
   *
   * @return a set with localids 
   */
  private HashSet<Integer> getAlternativeTextAds(HashSet<Integer> adgroups) {
    PreparedStatement select;
    HashSet<Integer> stack = new HashSet<Integer>();
    try {
      select = conn.prepareStatement(
          "SELECT DISTINCT ta.LocalId FROM TextAds ta, Schedule sc " +
          "WHERE sc.LocalId = ta.LocalId AND sc.AlternativeTextAd = 1 " +
          "AND LocalAdgroupId IN (" + getInClause(adgroups) + ")");
      ResultSet rs = select.executeQuery();

      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return stack;
  }
  
  /**
   * Gets the textAds' localids from DB that didn't match
   * the schedule rule defined in the destURL.
   *
   * @return a set with localids
   */
  private HashSet<Integer> getTextAdsOutSchedule() {
    PreparedStatement prodIdTextAds;
    HashSet<Integer> stack = new HashSet<Integer>();
    try {
      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE StartDate > -1 " +
          "AND StartDate > strftime('%Y%m%d', 'now')");
      ResultSet rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();

      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE enddate > -1 " +
          "AND enddate < strftime('%Y%m%d', 'now')");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
      
      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE StartHour > -1 " +
          "AND StartHour > strftime('%H','now')");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
      
      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE EndHour > -1 " +
          "AND EndHour < strftime('%H','now')");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();

      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE Sunday = 0 " +
          "AND strftime('%w','now') = '0'");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
      
      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE Monday = 0 " +
          "AND strftime('%w','now') = '1'");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();

      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE Tuesday = 0 " +
          "AND strftime('%w','now') = '2'");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
      
      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE Wednesday = 0 " +
          "AND strftime('%w','now') = '3'");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();

      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE Thursday = 0 " +
          "AND strftime('%w','now') = '4'");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();

      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE Friday = 0 " +
          "AND strftime('%w','now') = '5'");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();

      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE Saturday = 0 " +
          "AND strftime('%w','now') = '6'");
      rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return stack;
  }
  
  /**
   * Gets the localids representing the textads in DB with
   * any schedule rule defined at the destURL. At this point
   * it does not matter if the schedule is valid or not, i.e,
   * if the schedule points to activate the ads during the wednesday
   * and today is thursday, the localid for this textAds will be returned.
   *
   * @return a set with localids
   */
  private HashSet<Integer> getTextAdsWithSchedule() {
    PreparedStatement prodIdTextAds;
    HashSet<Integer> stack = new HashSet<Integer>();
    try {
      prodIdTextAds = conn.prepareStatement(
          "SELECT LocalId FROM Schedule WHERE StartDate <> -1 " +
          "OR enddate <> -1 OR StartHour <> -1 OR EndHour <> -1 " +
          "OR Sunday <> -1 OR Monday <> -1 OR Tuesday <> -1 " +
          "OR Wednesday <> -1 OR Thursday <> -1 OR Friday <> -1 " +
          "OR Saturday <> -1");
      ResultSet rs = prodIdTextAds.executeQuery();
      while (rs.next()) {
        stack.add(rs.getInt(1));
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return stack;
  }
  
  
  /**
   * Validate an AdParam with the rules defined in the destURL 
   *
   * @param adParam the adparam label to be used
   * @param adParamValue the adparam value
   * @param min the minimum value accepted by this adParam
   * @param max the maximum value accepted by this adParam
   * @return an empty string, case the AdParam don't match with the rules. 
   *         otherwise the adParamValue
   */
  public String validateAdParam(String adParam, String adParamValue, String min, 
      String max) {
    String sValue = "";
    if (adParam.toLowerCase().equals("price") || 
        adParam.toLowerCase().equals("regularprice") || 
        adParam.toLowerCase().equals("parcelvalue") ||
        adParam.toLowerCase().equals("lowerprice") ||
        adParam.toLowerCase().equals("lowerparcelprice")) {
      float value = Float.valueOf(adParamValue); 
      float adParamMin = Float.valueOf(min);
      float adParamMax = Float.valueOf(max);

      if (validAdParam(value, adParamMin, adParamMax)) {
        sValue = currency.toString(value);
      }
    } else if (adParam.toLowerCase().equals("parcel") || 
        adParam.toLowerCase().equals("units") || 
        adParam.toLowerCase().equals("discount") ||
        adParam.toLowerCase().equals("maxdiscount") ||
        adParam.toLowerCase().equals("items") || 
        adParam.toLowerCase().equals("lowerparcel")
        ) {
      int value = Integer.valueOf(adParamValue); 
      float adParamMin = Float.valueOf(min);
      float adParamMax = Float.valueOf(max);

      if (validAdParam(value, adParamMin, adParamMax)) {
        sValue = adParamValue;
      }
    }
    return sValue; 
  }
  
  /**
   * Validate an AdParam with the rules defined in the destURL 
   *
   * @param value the adparam value
   * @param min the minimum value accepted by this adParam
   * @param max the maximum value accepted by this adParam
   * @return true is the adParam match with the rules
   */
  public boolean validAdParam(float value, float min, float max) {
    if ((min > -1) && (value < min)) {
      return false;
    }
    if ((max > -1) && (value > max)) {
      return false;
    }    
    return true;
  }
  
  /**
   * Gets a set of localids representing the textads in the DB that
   * contains the aw_catid or aw_prodid tag in the textAd destURL
   *
   * @param adGroupsLocalId the adgroups localid that contains the taxtads
   *        to be validate
   * @param select the select SQL statement that will get the localids
   * @return a set with localids 
   */
  public HashSet<Integer> getTextAdsWithAdParam(int adGroupsLocalId, 
      PreparedStatement select) {
    HashSet<Integer> textAdIds = new HashSet<Integer>();  
    try {      
      select.setInt(1, adGroupsLocalId);
      ResultSet rs = select.executeQuery();
      while (rs.next()) {
        int textAdId = rs.getInt(1);
        textAdIds.add(textAdId);
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return textAdIds;
  }
  
  /**
   * Convert a set of localids into a IN clause to be used in a SQL 
   *
   * @param textAdIds a set with localids
   * @return the in clause
   */
  private String getInClause(HashSet<Integer> textAdIds) {
    String inClause = "";
    Iterator<Integer> i = textAdIds.iterator();
    while (i.hasNext()) {
      int id = i.next();
      inClause += Integer.toString(id);
      if (i.hasNext()) {
        inClause += ", ";
      }
    }
    return inClause;
  }
  
  /**
   * Updates all TextAds in the collection and set its status to Enable
   *
   * @param textAdIds list of localids representing the TextAds in DB
   */
  public void enableTextAds(HashSet<Integer> textAdIds) {
    try {
      PreparedStatement update = conn.prepareStatement(
          "UPDATE TextAds SET Status = 'ENABLED', IsTouched = 1, " +
          "IsUpdated = 1, IsBanned = 0 WHERE LocalId IN (" + 
          getInClause(textAdIds) + ") " + "AND Status = 'PAUSED'"
      );
      update.executeUpdate();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
  }
  
  /**
   * Gets the others text ads.
   *
   * @param adGroupLocalId the ad group local id
   * @param textAdIds the text ad ids
   * @return the others text ads
   */
  public HashSet<Integer> getOthersTextAds(int adGroupLocalId, 
      HashSet<Integer> textAdIds) {
    HashSet<Integer> othersTextAds = new HashSet<Integer>(); 
    try {
      PreparedStatement select = conn.prepareStatement(
          "SELECT LocalId FROM TextAds " +
          "WHERE LocalId NOT IN (" + getInClause(textAdIds) + ") " +
          "AND Status = 'ENABLED' AND LocalAdGroupId = ?"
      );
      ResultSet rs = select.executeQuery();
      while (rs.next()) {
        int localid = rs.getInt("LocalId");
        othersTextAds.add(localid);
      }      
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }
    return othersTextAds;
  }
   
  /**
   * Gets the ad param.
   *
   * @param adGroupLocalId the ad group local id
   * @return the ad param
   */
  public HashSet<KeywordParamStructure> getAdParam(int adGroupLocalId) {     
    PreparedStatement select;
    HashSet<KeywordParamStructure> kwdsParamStructure 
        = new HashSet<KeywordParamStructure>();
    try {
      select = conn.prepareStatement(
          "SELECT LocalId, KeywordId, AdgroupId, AdParam1, AdParam2 " +
          "FROM KeywordParams " +
          "WHERE LocalAdGroupId = ?");
      select.setInt(1, adGroupLocalId);
      ResultSet rs = select.executeQuery(); 
      while (rs.next()) {
        KeywordParamStructure kwdParamStructure = new KeywordParamStructure();
        kwdParamStructure.setLocalAdGroupId(adGroupLocalId);
        kwdParamStructure.setLocalId(rs.getInt("LocalId"));
        kwdParamStructure.setKeywordId(rs.getLong("KeywordId"));
        kwdParamStructure.setAdGroupId(rs.getLong("AdgroupId"));
        kwdParamStructure.setParam1(rs.getString("AdParam1"));
        kwdParamStructure.setParam2(rs.getString("AdParam2"));
        kwdsParamStructure.add(kwdParamStructure);        
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }
    return kwdsParamStructure;
  }
  
  /**
   * Gets the keyword ids.
   *
   * @param adGroupLocalId the ad group local id
   * @return the keyword ids
   */
  public HashSet<Integer> getKeywordIds(int adGroupLocalId) {
    HashSet<Integer> kwdIds = new HashSet<Integer>();
    
    try {
      PreparedStatement select = conn.prepareStatement(
          "SELECT LocalId FROM Keywords WHERE LocalAdGroupId = ? " +
          "AND IsNegative = 0"
      );
      select.setInt(1, adGroupLocalId);
      ResultSet rs = select.executeQuery();
      while (rs.next()) {
        int localId = rs.getInt("LocalId");
        kwdIds.add(localId);
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return kwdIds;
  }
  
  /**
   * Adds the ad param2 keywords.
   *
   * @param adGroupLocalId the ad group local id
   * @param kwdIds the kwd ids
   * @param adParam1 the ad param1
   * @param adParam2 the ad param2
   */
  private void addAdParam2Keywords(int adGroupLocalId, 
      HashSet<Integer> kwdIds, String adParam1, String adParam2) {
    try {
      PreparedStatement replace = conn.prepareStatement(
          "REPLACE INTO KeywordParams (LocalId, LocalAdgroupId, AdParam1, " +
          "AdParam2, IsUpdated) VALUES (?, ?, ?, ?, 1)"
      );
      
      Iterator<Integer> i = kwdIds.iterator();
      while (i.hasNext()) {
        int localId = i.next();
        replace.setInt(1, localId);
        replace.setInt(2, adGroupLocalId);
        replace.setString(3, adParam1);
        replace.setString(4, adParam2);
        replace.addBatch();
      }      
      replace.executeBatch();      
      
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
  }
  
  /**
   * Update ad param.
   *
   * @param adGroupLocalId the ad group local id
   * @param adParam1 the ad param1
   * @param adParam2 the ad param2
   */
  public void updateAdParam(int adGroupLocalId, String adParam1, 
      String adParam2) {
    try {      
      HashSet<KeywordParamStructure> kwdsParamStructure 
          = getAdParam(adGroupLocalId);
      HashSet<Integer> kwdIds = getKeywordIds(adGroupLocalId);
      
      // remove all keywords from adGroup that already has an adParam 
      // assigned to it 
      Iterator<KeywordParamStructure> i = kwdsParamStructure.iterator();
      while (i.hasNext()) {
        KeywordParamStructure kwdParamStructure = i.next();
        kwdIds.remove(kwdParamStructure.getLocalId());
      }
      
      // add to the new keywords the adParam values
      addAdParam2Keywords(adGroupLocalId, kwdIds, adParam1, adParam2);
            
      PreparedStatement updateParam1 = conn.prepareStatement(
          "UPDATE KeywordParams SET AdParam1 = ?, IsUpdated = 1 " +
          "WHERE LocalId = ? AND AdParam1 <> ?"
      );
      
      PreparedStatement updateParam2 = conn.prepareStatement(
          "UPDATE KeywordParams SET AdParam2 = ?, IsUpdated = 1 " +
          "WHERE LocalId = ? AND AdParam2 <> ?"
      );
      
      PreparedStatement updateBothParam = conn.prepareStatement(
          "UPDATE KeywordParams SET AdParam1 = ?, AdParam2 = ?, " +
          "IsUpdated = 1 WHERE LocalId = ? AND AdParam1 <> ? AND " +
          "AdParam2 <> ?"
      );
      
      i = kwdsParamStructure.iterator();
      while (i.hasNext()) {
        KeywordParamStructure kwdParamStructure = i.next();        
        if (!kwdParamStructure.getParam1().equals(adParam1) 
            && !kwdParamStructure.getParam2().equals(adParam2)) {
          updateBothParam.setString(1, adParam1);
          updateBothParam.setString(2, adParam2);
          updateBothParam.setInt(3, kwdParamStructure.getLocalId());
          updateBothParam.setString(4, adParam1);
          updateBothParam.setString(5, adParam2);
          updateBothParam.addBatch();
        } else if (!kwdParamStructure.getParam1().equals(adParam1)) {
          updateParam1.setString(1, adParam1);
          updateParam1.setInt(2, kwdParamStructure.getLocalId());
          updateParam1.setString(3, adParam1);
          updateParam1.addBatch();          
        } else if (!kwdParamStructure.getParam2().equals(adParam2)) {
          updateParam2.setString(1, adParam2);
          updateParam2.setInt(2, kwdParamStructure.getLocalId());
          updateParam2.setString(3, adParam2);
          updateParam2.addBatch();          
        }
      }
      updateBothParam.executeBatch();
      updateParam1.executeBatch();
      updateParam2.executeBatch();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }        
  }
  
  /**
   * Update ad param4 prod id.
   *
   * @return the hash set
   */
  public HashSet<Integer> updateAdParam4ProdId() {
    try {
      PreparedStatement select = conn.prepareStatement(
          "SELECT ca.*, ap.AdGroupLocalId, ap.AdParam1, ap.AdParam1Min, " +
          "ap.AdParam1Max,  ap.AdParam2, ap.AdParam2Min, ap.AdParam2Max " +
          "FROM AdGroup2ProdId ap, Catalog ca WHERE ca.ProdId = ap.ProdId"
      );
      
      PreparedStatement select2 = conn.prepareStatement(
          "SELECT ta.LocalId FROM TextAds ta, TextAd2ProdId tp " +
          "WHERE tp.LocalId = ta.LocalId " +
          "AND LocalAdgroupId = ?");
      
      return updateAdParam(select, select2);
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return new HashSet<Integer>();
  }
  
  /**
   * Update adParam for TextAds linked to a category of products and return 
   * is TextAds' localId.
   */
  private HashSet<Integer> updateAdParam4CatId() {    
    try {
      // first thing is to prepare the select statement to get the rules and
      // the AdParam values to be uptated
      PreparedStatement select = conn.prepareStatement(
          "SELECT ca.*, ac.AdGroupLocalId, ac.AdParam1, ac.AdParam1Min, " +
          "ac.AdParam1Max,  ac.AdParam2, ac.AdParam2Min, ac.AdParam2Max " +
          "FROM AdGroup2CatId ac, Category ca " +
          "WHERE ca.CategoryId = ac.CategoryId"
      );
      
      // get 
      PreparedStatement select2 = conn.prepareStatement(
          "SELECT ta.localId FROM TextAds ta, TextAd2CatId tp " +
          "WHERE tp.LocalId = ta.LocalId " +
          "AND LocalAdgroupId = ?");
      
      return updateAdParam(select, select2);
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }    
    return new HashSet<Integer>();
  }
  
  
  /**
   * Update ad param.
   *
   * @param select the select
   * @param select2 the select2
   * @return the hash set
   */
  public HashSet<Integer> updateAdParam(PreparedStatement select, 
      PreparedStatement select2) {
    HashSet<Integer> toBePaused = new HashSet<Integer>(); 
    try {
      ResultSet rs = select.executeQuery();
      while (rs.next()) {
        boolean moveOn = true;
        int adGroupLocalId = rs.getInt("AdGroupLocalId");
        String value1 = "";
        String value2 = "";
        
        String adParam1 = rs.getString("AdParam1");

        if (!adParam1.isEmpty()) {
          value1 = rs.getString(adParam1);
          String adParam1Min = rs.getString("AdParam1Min");
          String adParam1Max = rs.getString("AdParam1Max");
          value1 = validateAdParam(adParam1, value1, adParam1Min, adParam1Max);
          if (value1.isEmpty()) {
            moveOn = false;
          }
        }

        String adParam2 = rs.getString("AdParam2");

        if (!adParam2.isEmpty()) { 
          value2 = rs.getString(adParam2);
          String adParam2Min = rs.getString("AdParam2Min");
          String adParam2Max = rs.getString("AdParam2Max");
          value2 = validateAdParam(adParam2, value2, adParam2Min, adParam2Max);
          if (value2.isEmpty()) {
            moveOn = false;
          }
        }
        
        if (moveOn) {
          updateAdParam(adGroupLocalId, value1, value2);
          HashSet<Integer> textAdIds = getTextAdsWithAdParam(adGroupLocalId
              , select2);
          // make sure that textads without adParam will be paused
          toBePaused.addAll(getOthersTextAds(adGroupLocalId, textAdIds));
        } else {
          // clear all adParam values, just to make sure
          updateAdParam(adGroupLocalId, "", "");          
          HashSet<Integer> textAdIds = getTextAdsWithAdParam(adGroupLocalId
              , select2);
          // make sure that textads with adParam will be paused 
          toBePaused.addAll(textAdIds);
        }        
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    }   
    return toBePaused;
  }  
  
  /**
   * Loads EnhancedAds content from DB and do all updates needed, i.e.
   * Prices, TextAds Status, Schedules, etc.
   */
  public void updateTextAds() {
    log.info("Updating textAds status and adParam at DB");
    HashSet<Integer> textAdsToBeEnabled = new HashSet<Integer>();  
    HashSet<Integer> textAdsToBePaused = new HashSet<Integer>();

    // get textAds that referrer to products/categories in stock
    HashSet<Integer> textAdsInStock = getTextAdsInStock();
    
    // get textAds with any schedule rules
    HashSet<Integer> textAdsWithSchedule = getTextAdsWithSchedule();    
    
    // get textAds that referrer to prodId out of stock
    HashSet<Integer> textAdsOutOfStock = getUnAvailableTextAds();
    
    // get textAds that do not match with Schedule    
    HashSet<Integer> textAdsOutSchedule = getTextAdsOutSchedule();
    
    // get textAds that adParam do not match with min/max rules    
    HashSet<Integer> textAdsRestictedCatAdParam = updateAdParam4CatId();
    // get textAds that adParam do not match with min/max rules    
    HashSet<Integer> textAdsRestictedProdAdParam = updateAdParam4ProdId();
    
    textAdsToBeEnabled.addAll(textAdsInStock);
    textAdsToBeEnabled.addAll(textAdsWithSchedule);
    textAdsToBeEnabled.removeAll(textAdsOutOfStock);
    textAdsToBeEnabled.removeAll(textAdsOutSchedule);
    textAdsToBeEnabled.removeAll(textAdsRestictedCatAdParam);
    textAdsToBeEnabled.removeAll(textAdsRestictedProdAdParam);
    
    textAdsToBePaused.addAll(textAdsOutOfStock);
    textAdsToBePaused.addAll(textAdsOutSchedule);
    textAdsToBePaused.addAll(textAdsRestictedCatAdParam);
    textAdsToBePaused.addAll(textAdsRestictedProdAdParam);

    // HashSet<Integer> textAds2BeEnabled = getTextAds2BeEnabled(textAdsInStock);
    
    log.debug("Pausing TextAds linked to products/categories in stock");
    pauseTextAds(textAdsToBePaused);
    log.debug("Activating TextAds linked to products/categories in stock");
    enableTextAds(textAdsToBeEnabled);
    
    HashSet<Integer> adGroupsAlternative = getAdGroupsWithAlternativeTextAds();
    HashSet<Integer> adGroupsInStock = getAdGroupsWithTextAdsInStock();
    
    log.debug("Activating alternatives TextAds that should be activated");
    adGroupsAlternative.removeAll(adGroupsInStock);    
    HashSet<Integer> textAdsAlternative 
        = getAlternativeTextAds(adGroupsAlternative);    
    enableTextAds(textAdsAlternative);
    
    log.debug("Pausing alternatives TextAds that should be paused");
    textAdsAlternative = getAlternativeTextAds(adGroupsInStock);    
    pauseTextAds(textAdsAlternative);
    
    EmailReport emailReport = new EmailReport(conn, account, emailSettings);
    emailReport.createWorkbook(textAdsOutOfStock, textAdsOutSchedule, 
        textAdsInStock);
    emailReport.send();
  }
  
  /*
  public HashSet<Integer> getTextAds2BeEnabled(HashSet<Integer> textAdIds) {
    HashSet<Integer> localIds = new HashSet<Integer>();
    ResultSet rs = null;
    PreparedStatement prep = null;
    try {
      prep = conn.prepareStatement(
          "SELECT localId FROM TextAds WHERE LocalId IN (" + 
          getInClause(textAdIds) + ") " + "AND Status = 'PAUSED'"
      );
      rs = prep.executeQuery();
      while (rs.next()) {
        localIds.add(rs.getInt(1));
      }
      rs.close();
    } catch (SQLException e) {
      DumpExceptions.dumpSQLException(e);
    } finally {
      if (prep != null) {
        try {
          prep.close();
        } catch (SQLException e) {
          DumpExceptions.dumpSQLException(e);
        }
      }
      if (rs != null) {
        try {
          rs.close();
        } catch (SQLException e) {
          DumpExceptions.dumpSQLException(e);
        }
      }      
    }
    return localIds;
  }
  */
}
